package com.mjf.app.function;

import com.alibaba.fastjson.JSONObject;
import com.mjf.common.GmallConfig;
import com.mjf.utils.DimUtil;
import com.mjf.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 使用多线程的方式异步查询redis和HBase
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimAsyncJoinFunction<T>{

    private Connection conn;
    private ThreadPoolExecutor threadPoolExecutor;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    /**
     * 初始化 phoenix 连接及线程池
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        threadPoolExecutor = ThreadPoolUtil.getThreadPool();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    // 获取查询的主键
                    String id = getKey(input);
                    // 查询维度信息
                    JSONObject dimInfo = DimUtil.getDimInfo(conn, tableName, id);
                    // 补充维度信息
                    join(input, dimInfo);
                    // 将数据输出
                    resultFuture.complete(Collections.singletonList(input));
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        // 生产中可以在超时的时候重新发送一次请求
        System.out.println("TimeOut:" + input);
    }

}
